package com.ksyun.campus.client;

import com.ksyun.campus.client.domain.StatInfo;
import com.ksyun.campus.client.domain.BlockInfo;
import com.ksyun.campus.client.domain.ReplicaInfo;
import com.ksyun.campus.client.domain.ApiResponse;
import com.ksyun.campus.client.domain.OpenResponse;
import com.ksyun.campus.client.util.HttpClientUtil;
import com.fasterxml.jackson.core.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 文件系统输入流类
 * 继承自InputStream，提供从分布式文件系统读取文件数据的功能
 * 支持分块读取、副本容错、重试机制等特性
 * 
 * @author minfs
 * @version 1.0
 */
public class FSInputStream extends InputStream {
    private static final Logger logger = LoggerFactory.getLogger(FSInputStream.class);
    
    // 块大小常量：128MB = 128 * 1024 * 1024 = 134,217,728 bytes
    private static final long DEFAULT_BLOCK_SIZE = 533504L; // 128MB
    private static final int MAX_RETRY_ATTEMPTS = 3;
    
    private StatInfo statInfo;
    private String metaServerHost;
    private int metaServerPort;
    private String fileSystemName;
    
    // 将 currentOffset 改为 long 类型
    private long currentOffset;
    private long totalRead;
    
    // 文件块信息
    private List<BlockInfo> fileBlocks;
    private long blockSize;
    private String fileId;

    /**
     * 构造函数
     * 初始化文件输入流，并获取文件的块信息
     * 
     * @param statInfo 文件状态信息，包含文件路径、大小等
     * @param metaServerHost 元数据服务器主机地址
     * @param metaServerPort 元数据服务器端口
     * @param fileSystemName 文件系统名称
     * @throws IOException 当初始化文件块信息失败时抛出
     */
    public FSInputStream(StatInfo statInfo, String metaServerHost, int metaServerPort, String fileSystemName) throws IOException {
        this.statInfo = statInfo;
        this.metaServerHost = metaServerHost;
        this.metaServerPort = metaServerPort;
        this.fileSystemName = fileSystemName;
        this.currentOffset = 0;
        this.totalRead = 0;
        
        // 检查文件大小是否合理
        if (statInfo.getSize() < 0) {
            throw new IOException("Invalid file size: " + statInfo.getSize());
        }
        if (statInfo.getSize() > Long.MAX_VALUE / 2) {
            throw new IOException("File too large: size exceeds reasonable limit");
        }
        
        // 初始化文件块信息
        initializeFileBlocks();
    }
    
    /**
     * 初始化文件块信息
     * 从元数据服务器获取文件的块分布信息，包括块大小、副本位置等
     * 
     * @throws IOException 当获取或解析块信息失败时抛出
     */
    private void initializeFileBlocks() throws IOException {
        try {
            // 从 MetaServer 获取文件块信息
            String url = String.format("http://%s:%d/api/v1/meta/open", metaServerHost, metaServerPort);
            Map<String, String> headers = new HashMap<>();
            headers.put("fileSystemName", fileSystemName);
            headers.put("Content-Type", "application/json");
            
            String requestBody = String.format("{\"path\":\"%s\",\"mode\":\"READ\"}", statInfo.getPath());
            String response = HttpClientUtil.post(url, headers, requestBody);
            
            // 直接解析响应获取文件块信息
            ApiResponse<OpenResponse> apiResponse = HttpClientUtil.parseJson(response, 
                new TypeReference<ApiResponse<OpenResponse>>() {});
            
            if (!apiResponse.isSuccess()) {
                throw new IOException("API response failed: " + apiResponse.getMessage());
            }
            
            OpenResponse openResponse = apiResponse.getData();
            if (openResponse == null) {
                throw new IOException("Open response data is null");
            }
            
            this.fileId = openResponse.getFileId();
            this.blockSize = openResponse.getBlockSize();
            this.fileBlocks = openResponse.getBlocks();
            
            // 如果块大小为0或负数，使用默认值
            if (this.blockSize <= 0) {
                this.blockSize = DEFAULT_BLOCK_SIZE;
                logger.warn("Invalid block size in response, using default: {} bytes", DEFAULT_BLOCK_SIZE);
            }
            
            logger.info("File blocks initialized - fileId: {}, blockSize: {} bytes ({}MB), blocks: {}", 
                       fileId, blockSize, blockSize / (1024 * 1024), fileBlocks != null ? fileBlocks.size() : 0);
            
            // 验证块信息的完整性
            if (fileBlocks == null || fileBlocks.isEmpty()) {
                throw new IOException("File blocks validation failed: blocks are null or empty");
            }
            
            // 过滤并验证块信息 - 只保留有有效副本的块
            List<BlockInfo> validBlocks = new ArrayList<>();
            for (int i = 0; i < fileBlocks.size(); i++) {
                BlockInfo block = fileBlocks.get(i);
                if (block == null) {
                    logger.warn("Block at index {} is null, skipping", i);
                    continue;
                }
                
                if (block.getReplicas() == null || block.getReplicas().isEmpty()) {
                    logger.warn("Block {} has no replicas, skipping", block.getBlkIdx());
                    continue;
                }
                
                // 验证副本信息的有效性
                boolean hasValidReplica = false;
                for (ReplicaInfo replica : block.getReplicas()) {
                    if (replica.getHost() != null && !replica.getHost().isEmpty() && replica.getPort() > 0) {
                        hasValidReplica = true;
                        break;
                    }
                }
                
                if (hasValidReplica) {
                    validBlocks.add(block);
                    logger.debug("Block {} is valid with {} replicas", block.getBlkIdx(), block.getReplicas().size());
                } else {
                    logger.warn("Block {} has no valid replicas, skipping", block.getBlkIdx());
                }
            }
            
            // 更新为过滤后的有效块列表
            this.fileBlocks = validBlocks;
            
            if (this.fileBlocks.isEmpty()) {
                throw new IOException("No valid blocks found for file " + statInfo.getPath());
            }
            
            logger.debug("File blocks validation passed: {} blocks", fileBlocks.size());
            
        } catch (Exception e) {
            logger.error("Failed to initialize file blocks", e);
            throw new IOException("Failed to initialize file blocks", e);
        }
    }
    

    /**
     * 读取单个字节
     * 从文件当前位置读取一个字节，并推进读取位置
     * 
     * @return 读取的字节值（0-255），如果到达文件末尾返回-1
     * @throws IOException 当读取失败时抛出
     */
    @Override
    public int read() throws IOException {
        if (currentOffset >= statInfo.getSize()) {
            return -1; // EOF
        }
        
        // 读取单个字节
        byte[] singleByte = new byte[1];
        int bytesRead = read(singleByte, 0, 1);
        if (bytesRead == -1) {
            return -1;
        }
        return singleByte[0] & 0xFF;
    }

    /**
     * 读取字节数组
     * 从文件当前位置读取指定长度的字节到数组中
     * 
     * @param b 存储读取数据的字节数组
     * @return 实际读取的字节数，如果到达文件末尾返回-1
     * @throws IOException 当读取失败时抛出
     */
    @Override
    public int read(byte[] b) throws IOException {
        return read(b, 0, b.length);
    }

    /**
     * 读取字节数组的指定部分
     * 从文件当前位置读取指定长度的字节到数组的指定位置
     * 
     * @param b 存储读取数据的字节数组
     * @param off 数组中的起始偏移量
     * @param len 要读取的最大字节数
     * @return 实际读取的字节数，如果到达文件末尾返回-1
     * @throws IOException 当读取失败时抛出
     */
    @Override
    public int read(byte[] b, int off, int len) throws IOException {
        if (currentOffset >= statInfo.getSize()) {
            return -1; // EOF
        }
        
        if (len <= 0) {
            return 0;
        }
        
        // 如果文件块信息不可用，直接返回错误
        if (fileBlocks == null || fileBlocks.isEmpty()) {
            throw new IOException("File blocks information not available - cannot read file");
        }
        
        // 修复文件大小计算
        long remainingInFile = statInfo.getSize() - currentOffset;
        if (remainingInFile > Integer.MAX_VALUE) {
            throw new IOException("File too large: remaining bytes exceed int range");
        }
        int toRead = (int) Math.min(len, remainingInFile);
        
        if (toRead <= 0) {
            return -1;
        }
        
        // 根据当前偏移量确定需要读取的块
        int bytesRead = 0;
        int remainingToRead = toRead;
        int currentPos = off;
        
        while (remainingToRead > 0 && currentOffset < statInfo.getSize()) {
            // 计算当前偏移量对应的块索引和块内偏移量
            BlockLocation blockLocation = getBlockLocation(currentOffset);
            if (blockLocation == null) {
                logger.error("Failed to get block location for offset: {}", currentOffset);
                break;
            }
            
            // 计算在当前块中可以读取的字节数
            // 使用实际的块大小，而不是计算出的块大小
            long actualBlockSize = blockLocation.block.getSize();
            if (actualBlockSize <= 0) {
                actualBlockSize = blockSize; // 如果块大小无效，使用默认块大小
            }
            
            // 添加边界检查，确保不会溢出
            long availableInBlock = actualBlockSize - blockLocation.offsetInBlock;
            if (availableInBlock > Integer.MAX_VALUE) {
                throw new IOException("Block too large: available bytes exceed int range");
            }
            
            int bytesInCurrentBlock = Math.min(remainingToRead, (int) availableInBlock);
            
            // 从当前块读取数据，支持重试
            byte[] blockData = readFromBlockWithRetry(blockLocation, bytesInCurrentBlock);
            if (blockData == null || blockData.length == 0) {
                logger.error("Failed to read from block {} after retries", 
                           blockLocation.block.getBlkIdx());
                break;
            }
            
            // 复制数据到目标数组
            System.arraycopy(blockData, 0, b, currentPos, blockData.length);
            currentPos += blockData.length;
            currentOffset += blockData.length;
            totalRead += blockData.length;
            bytesRead += blockData.length;
            remainingToRead -= blockData.length;
            
            logger.debug("Read {} bytes from block {}, current offset: {}, remaining: {}", 
                       blockData.length, blockLocation.block.getBlkIdx(), 
                       currentOffset, remainingToRead);
        }
        
        return bytesRead > 0 ? bytesRead : -1;
    }
    
    /**
     * 根据文件偏移量计算对应的块位置
     * 计算指定文件偏移量对应的块索引、块内偏移量和块大小
     * 
     * @param fileOffset 文件中的偏移量
     * @return 块位置信息对象，如果计算失败返回null
     * @throws IOException 当计算过程中发生异常时抛出
     */
    private BlockLocation getBlockLocation(long fileOffset) throws IOException {
        if (fileBlocks == null || fileBlocks.isEmpty()) {
            logger.error("File blocks are null or empty");
            return null;
        }
        
        if (fileOffset < 0) {
            logger.error("Invalid file offset: {}", fileOffset);
            return null;
        }
        
        // 按顺序查找包含该偏移量的块
        long currentOffset = 0;
        for (int i = 0; i < fileBlocks.size(); i++) {
            BlockInfo block = fileBlocks.get(i);
            if (block == null) {
                logger.error("Block at index {} is null", i);
                continue;
            }
            
            long actualBlockSize = block.getSize();
            if (actualBlockSize <= 0) {
                actualBlockSize = blockSize; // 使用默认块大小
            }
            
            // 检查文件偏移量是否在这个块中
            if (fileOffset >= currentOffset && fileOffset < currentOffset + actualBlockSize) {
                long offsetInBlockLong = fileOffset - currentOffset;
                if (offsetInBlockLong > Integer.MAX_VALUE) {
                    throw new IOException("Offset in block too large: exceeds int range");
                }
                int offsetInBlock = (int) offsetInBlockLong;
                
                logger.debug("Block location found - offset: {}, blockListIndex: {}, blockIdx: {}, offsetInBlock: {}, blockSize: {}", 
                           fileOffset, i, block.getBlkIdx(), offsetInBlock, actualBlockSize);
                
                return new BlockLocation(block, offsetInBlock, actualBlockSize);
            }
            
            currentOffset += actualBlockSize;
        }
        
        logger.error("File offset {} is beyond file size (total valid blocks: {}, total size: {})", 
                   fileOffset, fileBlocks.size(), currentOffset);
        return null;
    }
    
    /**
     * 从指定块读取数据，支持重试机制
     * 当读取失败时，会尝试多次重试，提高读取成功率
     * 
     * @param blockLocation 块位置信息
     * @param length 要读取的字节数
     * @return 读取的数据字节数组，如果所有重试都失败返回null
     * @throws IOException 当所有重试都失败时抛出
     */
    private byte[] readFromBlockWithRetry(BlockLocation blockLocation, int length) throws IOException {
        Exception lastException = null;
        
        for (int attempt = 1; attempt <= MAX_RETRY_ATTEMPTS; attempt++) {
            try {
                byte[] data = readFromBlock(blockLocation, length);
                if (data != null && data.length > 0) {
                    if (attempt > 1) {
                        logger.info("Successfully read from block {} on attempt {}", 
                                   blockLocation.block.getBlkIdx(), attempt);
                    }
                    return data;
                }
            } catch (Exception e) {
                lastException = e;
                logger.warn("Attempt {} failed to read from block {}: {}", 
                           attempt, blockLocation.block.getBlkIdx(), e.getMessage());
                
                if (attempt < MAX_RETRY_ATTEMPTS) {
                    try {
                        // 指数退避重试
                        Thread.sleep(100 * attempt);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new IOException("Read operation interrupted", ie);
                    }
                }
            }
        }
        
        // 所有重试都失败
        if (lastException != null) {
            throw new IOException("Failed to read from block " + blockLocation.block.getBlkIdx() + 
                                " after " + MAX_RETRY_ATTEMPTS + " attempts", lastException);
        }
        
        return null;
    }
    
    /**
     * 从指定块读取数据 - 完全使用块级读取API
     * 尝试从块的多个副本中读取数据，实现容错机制
     * 
     * @param blockLocation 块位置信息
     * @param length 要读取的字节数
     * @return 读取的数据字节数组，如果所有副本都失败返回null
     * @throws IOException 当读取过程中发生异常时抛出
     */
    private byte[] readFromBlock(BlockLocation blockLocation, int length) throws IOException {
        BlockInfo block = blockLocation.block;
        
        // 尝试从块的副本中读取数据
        for (ReplicaInfo replica : block.getReplicas()) {
            try {
                // 使用块级读取API
                byte[] blockData = readBlockFromReplica(replica, block);
                if (blockData != null && blockData.length > 0) {
                    // 从完整块数据中提取所需的部分
                    int startOffset = blockLocation.offsetInBlock;
                    int actualLength = Math.min(length, blockData.length - startOffset);
                    
                    if (startOffset < blockData.length && actualLength > 0) {
                        byte[] result = new byte[actualLength];
                        System.arraycopy(blockData, startOffset, result, 0, actualLength);
                        
                        logger.debug("Successfully read {} bytes from block API {}:{} (block {}, offset: {})", 
                                   actualLength, replica.getHost(), replica.getPort(), block.getBlkIdx(), startOffset);
                        return result;
                    }
                } else {
                    logger.warn("Empty response from replica {}:{} for block {}", 
                               replica.getHost(), replica.getPort(), block.getBlkIdx());
                }
            } catch (Exception e) {
                logger.warn("Failed to read from replica {}:{} for block {} - {}", 
                    replica.getHost(), replica.getPort(), block.getBlkIdx(), e.getMessage());
            }
        }
        
        // 所有副本都失败
        logger.error("All replicas failed for block {}", block.getBlkIdx());
        return null;
    }
    

    
    /**
     * 从副本读取完整块数据
     */
    private byte[] readBlockFromReplica(ReplicaInfo replica, BlockInfo block) throws IOException {
        try {
            String url = String.format("http://%s:%d/api/v1/data/block?fileId=%s&blkIdx=%d",
                replica.getHost(), replica.getPort(), fileId, block.getBlkIdx());
            
            Map<String, String> headers = new HashMap<>();
            headers.put("Accept", "application/octet-stream");
            
            byte[] data = HttpClientUtil.getBytes(url, headers);
            if (data != null) {
                logger.debug("Successfully read block data from {}:{} (fileId: {}, blkIdx: {}, size: {})", 
                           replica.getHost(), replica.getPort(), fileId, block.getBlkIdx(), data.length);
                return data;
            }
            
            return null;
        } catch (Exception e) {
            logger.warn("Block-level read failed from {}:{} - {}", 
                       replica.getHost(), replica.getPort(), e.getMessage());
            throw new IOException("Block-level read failed", e);
        }
    }

    /**
     * 关闭输入流
     * 清理资源并重置状态
     * 
     * @throws IOException 当关闭过程中发生异常时抛出
     */
    @Override
    public void close() throws IOException {
        try {
            logger.debug("Closing FSInputStream for file: {}, total bytes read: {}", 
                       statInfo != null ? statInfo.getPath() : "unknown", totalRead);
        } finally {
            // 重置状态
            currentOffset = 0;
            totalRead = 0;
            fileBlocks = null;
            super.close();
        }
    }
    


    /**
     * 块位置信息内部类
     * 封装了块信息、块内偏移量和块大小
     */
    private static class BlockLocation {
        final BlockInfo block;
        final int offsetInBlock;
        final long blockSize;
        
        /**
         * 构造函数
         * 
         * @param block 块信息
         * @param offsetInBlock 块内偏移量
         * @param blockSize 块大小
         */
        BlockLocation(BlockInfo block, int offsetInBlock, long blockSize) {
            this.block = block;
            this.offsetInBlock = offsetInBlock;
            this.blockSize = blockSize;
        }
    }
}
